/**
 * Simple Apache Kafka producer
 * using the Kafka driver from librdkafka
 * (https://github.com/edenhill/librdkafka)
 */

#ifdef USE_RDKAFKA

#include "sys.h"
#include "log.h"

/* Typical include path would be <librdkafka/rdkafka.h>, but this program
 * is builtin from within the librdkafka source tree and thus differs. */
#include "librdkafka/rdkafka.h"

static int run = 1;

/**
 * @brief Signal termination of program
 */
// static void stop(int sig)
// {
//     run = 0;
//     fclose(stdin); /* abort fgets() */
// }

/**
 * @brief Message delivery report callback.
 *
 * This callback is called exactly once per message, indicating if
 * the message was succesfully delivered
 * (rkmessage->err == RD_KAFKA_RESP_ERR_NO_ERROR) or permanently
 * failed delivery (rkmessage->err != RD_KAFKA_RESP_ERR_NO_ERROR).
 *
 * The callback is triggered from rd_kafka_poll() and executes on
 * the application's thread.
 */
static void dr_msg_cb(rd_kafka_t* rk,
    const rd_kafka_message_t* rkmessage, void* opaque)
{
    if (rkmessage->err)
        log_error("Kafka %% Message delivery failed: %s",
            rd_kafka_err2str(rkmessage->err));
    else
        log_trace(
            "Kafka %% Message delivered (%zd bytes, "
            "partition %" PRId32 ")",
            rkmessage->len, rkmessage->partition);

    /* The rkmessage is destroyed automatically by librdkafka */
}

static int _kafka_send(char* buf, int len, char* brokers, char* topic)
{
    rd_kafka_t* rk; /* Producer instance handle */
    rd_kafka_topic_t* rkt; /* Topic object */
    rd_kafka_conf_t* conf; /* Temporary configuration object */
    char errstr[512]; /* librdkafka API error reporting buffer */
    //     char buf[512]; /* Message value temporary buffer */
    //     const char* brokers; /* Argument: broker list */
    //     const char* topic; /* Argument: topic to produce to */
    int ret = 0;//0:success, 1:failed
    /*
         * Argument validation
         */
    //     if (argc != 3) {
    //         fprintf(stderr, "%% Usage: %s <broker> <topic>\n", argv[0]);
    //         return 1;
    //     }

    //     brokers = argv[1];
    //     topic = argv[2];
    /*
         * Create Kafka client configuration place-holder
         */
    conf = rd_kafka_conf_new();

    /* Set bootstrap broker(s) as a comma-separated list of
         * host or host:port (default port 9092).
         * librdkafka will use the bootstrap brokers to acquire the full
         * set of brokers from the cluster. */
    if (rd_kafka_conf_set(conf, "bootstrap.servers", brokers,
            errstr, sizeof(errstr))
        != RD_KAFKA_CONF_OK) {
        log_error("%s\n", errstr);
        return 1;
    }

    /* Set the delivery report callback.
         * This callback will be called once per message to inform
         * the application if delivery succeeded or failed.
         * See dr_msg_cb() above. */
    rd_kafka_conf_set_dr_msg_cb(conf, dr_msg_cb);

    /*
         * Create producer instance.
         *
         * NOTE: rd_kafka_new() takes ownership of the conf object
         *       and the application must not reference it again after
         *       this call.
         */
    rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr));
    if (!rk) {
        log_error(
            "%% Failed to create new producer: %s\n", errstr);
        return 1;
    }

    /* Create topic object that will be reused for each message
         * produced.
         *
         * Both the producer instance (rd_kafka_t) and topic objects (topic_t)
         * are long-lived objects that should be reused as much as possible.
         */
    rkt = rd_kafka_topic_new(rk, topic, NULL);
    if (!rkt) {
        log_error("%% Failed to create topic object: %s\n",
            rd_kafka_err2str(rd_kafka_last_error()));
        rd_kafka_destroy(rk);
        return 1;
    }

    /* Signal handler for clean shutdown */
    //     signal(SIGINT, stop);

    // log_error(
    //     "%% Type some text and hit enter to produce message\n"
    //     "%% Or just hit enter to only serve delivery reports\n"
    //     "%% Press Ctrl-C or Ctrl-D to exit\n");

    //     while (run && fgets(buf, sizeof(buf), stdin)) {
    //         size_t len = strlen(buf);

    // if (buf[len - 1] == '\n') /* Remove newline */
    //     buf[--len] = '\0';

    // if (len == 0) {
    //     /* Empty line: only serve delivery reports */
    //     rd_kafka_poll(rk, 0 /*non-blocking */);
    //     continue;
    // }

    /*
                 * Send/Produce message.
                 * This is an asynchronous call, on success it will only
                 * enqueue the message on the internal producer queue.
                 * The actual delivery attempts to the broker are handled
                 * by background threads.
                 * The previously registered delivery report callback
                 * (dr_msg_cb) is used to signal back to the application
                 * when the message has been delivered (or failed).
                 */
    // retry:
    if (rd_kafka_produce(
            /* Topic object */
            rkt,
            /* Use builtin partitioner to select partition*/
            RD_KAFKA_PARTITION_UA,
            /* Make a copy of the payload. */
            RD_KAFKA_MSG_F_COPY,
            /* Message payload (value) and length */
            buf, len,
            /* Optional key and its length */
            NULL, 0,
            /* Message opaque, provided in
                             * delivery report callback as
                             * msg_opaque. */
            NULL)
        == -1) {
        /**
                         * Failed to *enqueue* message for producing.
                         */
        log_warn(
            "%% Failed to produce to topic %s: %s\n",
            rd_kafka_topic_name(rkt),
            rd_kafka_err2str(rd_kafka_last_error()));

        /* Poll to handle delivery reports */
        if (rd_kafka_last_error() == RD_KAFKA_RESP_ERR__QUEUE_FULL) {
            /* If the internal queue is full, wait for
                                 * messages to be delivered and then retry.
                                 * The internal queue represents both
                                 * messages to be sent and messages that have
                                 * been sent or failed, awaiting their
                                 * delivery report callback to be called.
                                 *
                                 * The internal queue is limited by the
                                 * configuration property
                                 * queue.buffering.max.messages */
            rd_kafka_poll(rk, 1000 /*block for max 1000ms*/);
            // goto retry;
            ret = 1;
        }
    } else {
        log_trace("Kafka %% Enqueued message (%zd bytes) "
                  "for topic %s",
            len, rd_kafka_topic_name(rkt));
    }

    /* A producer application should continually serve
                 * the delivery report queue by calling rd_kafka_poll()
                 * at frequent intervals.
                 * Either put the poll call in your main loop, or in a
                 * dedicated thread, or call it after every
                 * rd_kafka_produce() call.
                 * Just make sure that rd_kafka_poll() is still called
                 * during periods where you are not producing any messages
                 * to make sure previously produced messages have their
                 * delivery report callback served (and any other callbacks
                 * you register). */
    // rd_kafka_poll(rk, 0 /*non-blocking*/);
    rd_kafka_poll(rk, 1000 );
    //     }

    /* Wait for final messages to be delivered or fail.
         * rd_kafka_flush() is an abstraction over rd_kafka_poll() which
         * waits for all messages to be delivered. */
    log_trace( "%% Flushing final messages..");
    rd_kafka_flush(rk, 10 * 1000 /* wait for max 10 seconds */);
    log_trace("");
    /* Destroy topic object */
    rd_kafka_topic_destroy(rkt);

    /* Destroy the producer instance */
    rd_kafka_destroy(rk);

    return ret;
}

int kafka_send( char* brokers, char* topic, char* buf, int len)
{
    int ret = 0;
    char *broker, *delim = ",", *saveptr;
    broker = strtok_r(brokers, delim, &saveptr);
    while (broker) {
        log_trace("kafak broker:%s", broker);
        if (_kafka_send(buf, len, brokers, topic) == 0) {
            ret = 1;
            break;
        }
        broker = strtok_r(NULL, delim, &saveptr);
    }
    return ret;
}

#endif // USE_RDKAFKA